package com.ftg.learn.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ftg.learn.entity.CanalMessageEntity;
import com.ftg.learn.entity.SysRoleEntity;
import com.ftg.learn.repository.EsNewsRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Slf4j
public class CanalConsumer {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private EsNewsRepository exampleRepository;

    public CanalConsumer(EsNewsRepository exampleRepository) {
        this.exampleRepository = exampleRepository;
    }

    @KafkaListener(topics = "canal_db_example", groupId = "canal-es-sync-group")
    public void listen(String message) {
        try {
            log.info("Received message: {}", message);
            CanalMessageEntity canalMsg = objectMapper.readValue(message, CanalMessageEntity.class);

            if (!"wensheng".equals(canalMsg.getDatabase())) {
                return;
            }

            String type = canalMsg.getType();
            for (Map<String, String> row : canalMsg.getData()) {

                switch (type) {
                    case "INSERT":
                        SysRoleEntity entity = buildRoleEntity(row);
                        exampleRepository.save(entity);
                        break;
                    case "UPDATE":
                        exampleRepository.save(buildRoleEntity(row));
                        break;
                    case "DELETE":
                        String idStr = row.get("id");
                        if (idStr != null) {
                            Integer id = Integer.valueOf(idStr);  // 假设 ID 是整数类型
                            exampleRepository.deleteById(Long.valueOf(id));
                        } else {
                            log.warn("ID is missing in DELETE event data: {}", row);
                        }
                        break;
                    default:
                        log.warn("Unsupported event type: {}", type);
                }
            }
        } catch (Exception e) {
            log.error("Error processing Kafka message", e);
        }
    }

    private SysRoleEntity buildRoleEntity(Map<String, String> row) {
        SysRoleEntity example = new SysRoleEntity();
        example.setId((int) Long.parseLong(row.get("id")));
        example.setRoleName(row.get("role_name"));
        example.setRoleCode(row.get("role_code"));
        example.setCreateTime(row.get("create_time"));
        example.setUpdateTime(row.get("update_time"));
        return example;
    }
}
